package com.demo.attC;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Spliterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class ForkingStreamConsumer<T> implements Consumer<T>, Results {

	static final Object END_OF_STREAM = new Object();

    private final List<BlockingQueue<T>> queues;
    private final Map<Object, Future<?>> actions;

    ForkingStreamConsumer(List<BlockingQueue<T>> queues, Map<Object, Future<?>> actions) {
        this.queues = queues;
        this.actions = actions;
    }

    @Override
    public void accept(T t) {
        queues.forEach(q -> q.add(t));
        System.out.println(queues);
    }

    @Override
    public <R> R get(Object key) {
        try {
            return ((Future<R>) actions.get(key)).get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    void finish() {
        accept((T) END_OF_STREAM);
    }

    public static <T> ForkingStreamConsumer<T> build(Map<Object, Function<Stream<T>, ?>> forks) {
    	//创建由队列组成的列表，每一个队列对应一个操作
    	//BlockingQueue：当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。
    	//当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。
        List<BlockingQueue<T>> queues = new ArrayList<>();

        //建立用于标识操作的键与包含操作结果的Future之间的映射关系
        Map<Object, Future<?>> actions =
                forks.entrySet().stream().reduce(
                        new HashMap<Object, Future<?>>(),
                        (map, e) -> {
                            map.put(e.getKey(),
                                    getOperationResult(queues, e.getValue()));
                            return map;
                        },
                        (m1, m2) -> {
                            m1.putAll(m2);
                            return m1;
                        });

        return new ForkingStreamConsumer<>(queues, actions);
    }

    private static <T> Future<?> getOperationResult(List<BlockingQueue<T>> queues, Function<Stream<T>, ?> f) {
        BlockingQueue<T> queue = new LinkedBlockingQueue<>();
        //创建一个队列，并将其添加到队列的列表中
        queues.add(queue);
        //创建一个Spliterator，遍历队列中的元素
        Spliterator<T> spliterator = new BlockingQueueSpliterator<>(queue);
        //创建一个流，将Spliterator作为数据源
        Stream<T> source = StreamSupport.stream(spliterator, false);
        //创建一个Future对象，以异步的方式计算在流上执行特定函数的结果
        System.out.println("创建Future");
        CompletableFuture<?> fu = CompletableFuture.supplyAsync( () -> f.apply(source) );
        System.out.println("创建完Future");
        return fu;
    }

}
